/*
 * (C) Copyright 2014-2016, by Dimitrios Michail
 *
 * JHeaps Library
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.jheaps.tree

import org.jheaps.AddressableHeap
import org.jheaps.MergeableAddressableHeap
import org.jheaps.annotations.ConstantTime
import org.jheaps.annotations.LogLogTime
import org.jheaps.annotations.LogarithmicTime
import java.io.Serializable
import java.util.*
import kotlin.Comparator

/**
 * The costless meld variant of the pairing heaps. The heap is sorted according
 * to the [natural ordering][Comparable] of its keys, or by a
 * [Comparator] provided at heap creation time, depending on which
 * constructor is used.
 *
 *
 *
 * This implementation provides amortized O(1) time for `findMin` and
 * `insert`, amortized O(log(n)) for `deleteMin` and `delete`
 * and amortized O(loglog(n)) for the `decreaseKey` operation. The
 * operation `meld` takes amortized zero time.
 *
 *
 *
 * This variant of the pairing heap is due to Amr Elmasry, described in detail
 * in the following
 * [paper](http://dx.doi.org/10.1007/978-3-642-15781-3_16):
 *
 *  * Amr Elmasry, Pairing Heaps with Costless Meld, In Proceedings of the 18th
 * Annual European Symposium on Algorithms (ESA 2010), 183--193, 2010.
 *
 *
 *
 *
 * All the above bounds, however, assume that the user does not perform
 * cascading melds on heaps such as:
 *
 * <pre>
 * d.meld(e);
 * c.meld(d);
 * b.meld(c);
 * a.meld(b);
</pre> *
 *
 * The above scenario, although efficiently supported by using union-find with
 * path compression, invalidates the claimed bounds.
 *
 *
 *
 * Note that the ordering maintained by a pairing heap, like any heap, and
 * whether or not an explicit comparator is provided, must be *consistent
 * with `equals`* if this heap is to correctly implement the
 * `AddressableHeap` interface. (See `Comparable` or
 * `Comparator` for a precise definition of *consistent with
 * equals*.) This is so because the `AddressableHeap` interface is
 * defined in terms of the `equals` operation, but a pairing heap performs
 * all key comparisons using its `compareTo` (or `compare`) method,
 * so two keys that are deemed equal by this method are, from the standpoint of
 * the pairing heap, equal. The behavior of a heap *is* well-defined even
 * if its ordering is inconsistent with `equals`; it just fails to obey
 * the general contract of the `AddressableHeap` interface.
 *
 *
 *
 * **Note that this implementation is not synchronized.** If
 * multiple threads access a heap concurrently, and at least one of the threads
 * modifies the heap structurally, it *must* be synchronized externally.
 * (A structural modification is any operation that adds or deletes one or more
 * elements or changing the key of some element.) This is typically accomplished
 * by synchronizing on some object that naturally encapsulates the heap.
 *
 * @param <K>
 * the type of keys maintained by this heap
 * @param <V>
 * the type of values maintained by this heap
 *
 * @author Dimitrios Michail
 *
 * @see PairingHeap
 *
 * @see FibonacciHeap
</V></K> */
class CostlessMeldPairingHeap<K, V> @ConstantTime constructor(comparator: Comparator<in K?>?) :
    MergeableAddressableHeap<K, V>, Serializable {
    /**
     * The comparator used to maintain order in this heap, or null if it uses
     * the natural ordering of its keys.
     *
     * @serial
     */
    private val comparator: Comparator<in K?>?

    /**
     * The root of the pairing heap
     */
    private var root: Node? = null

    /**
     * Size of the pairing heap
     */
    private var size: Long = 0

    /**
     * The decrease pool
     */
    private var decreasePool: Array<Node?>

    /**
     * How many elements are valid in the decrease pool
     */
    private var decreasePoolSize: Byte = 0

    /**
     * Index of node with minimum key in the decrease pool. Not existent if
     * decreasePoolMin &gt;= decreasePoolSize.
     */
    private var decreasePoolMinPos: Byte = 0

    /**
     * Comparator for nodes in the decrease pool. Initialized lazily and used
     * when sorting entries in the decrease pool.
     */
    @Transient
    private var decreasePoolComparator: Comparator<Node?>?

    /**
     * Used to reference the current heap or some other pairing heap in case of
     * melding, so that handles remain valid even after a meld, without having
     * to iterate over them.
     *
     * In order to avoid maintaining a full-fledged union-find data structure,
     * we disallow a heap to be used in melding more than once. We use however,
     * path-compression in case of cascading melds, that it, a handle moves from
     * one heap to another and then another.
     */
    private var other: CostlessMeldPairingHeap<K, V>

    /**
     * Constructs a new, empty heap, using the natural ordering of its keys. All
     * keys inserted into the heap must implement the [Comparable]
     * interface. Furthermore, all such keys must be *mutually
     * comparable*: `k1.compareTo(k2)` must not throw a
     * `ClassCastException` for any keys `k1` and `k2` in the
     * heap. If the user attempts to put a key into the heap that violates this
     * constraint (for example, the user attempts to put a string key into a
     * heap whose keys are integers), the `insert(Object key)` call will
     * throw a `ClassCastException`.
     */
    @ConstantTime
    constructor() : this(null)

    /**
     * Constructs a new, empty heap, ordered according to the given comparator.
     * All keys inserted into the heap must be *mutually comparable* by
     * the given comparator: `comparator.compare(k1,
     * k2)` must not throw a `ClassCastException` for any keys `k1`
     * and `k2` in the heap. If the user attempts to put a key into the
     * heap that violates this constraint, the `insert(Object key)` call
     * will throw a `ClassCastException`.
     *
     * @param comparator
     * the comparator that will be used to order this heap. If
     * `null`, the [natural ordering][Comparable] of
     * the keys will be used.
     */
    init {
        this.decreasePool = java.lang.reflect.Array.newInstance(Node::class.java, DEFAULT_DECREASE_POOL_SIZE) as Array<Node?>

        this.comparator = comparator
        this.decreasePoolComparator = null
        this.other = this
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @LogarithmicTime(amortized = true)
    override fun insert(key: K, value: V?): Node {
        check(other === this) { "A heap cannot be used after a meld" }
        if (key == null) {
            throw NullPointerException("Null keys not permitted")
        }
        val n = Node(this, key, value)
        root = if (comparator == null) {
            link(root, n)
        } else {
            linkWithComparator(root, n)
        }
        size++
        return n
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @LogarithmicTime(amortized = true)
    override fun insert(key: K): Node {
        return insert(key, null)
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun findMin(): Node{
        if (size == 0L) {
            throw NoSuchElementException()
        } else if (decreasePoolMinPos >= decreasePoolSize) {
            return root!!
        } else {
            val poolMin = decreasePool[decreasePoolMinPos.toInt()]
            val c: Int = if (comparator == null) {
                (root!!.key as Comparable<K>?)!!.compareTo(poolMin!!.key)
            } else {
                comparator.compare(root!!.key, poolMin!!.key)
            }
            return if (c <= 0) {
                root!!
            } else {
                poolMin
            }
        }
    }

    @get:ConstantTime
    override val isEmpty: Boolean
        get() = size == 0L

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun size(): Long {
        return size
    }

    /**
     * {@inheritDoc}
     */
    override fun comparator(): Comparator<in K?>? {
        return comparator
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun clear() {
        root = null
        size = 0
        decreasePool = arrayOfNulls(DEFAULT_DECREASE_POOL_SIZE)
            //java.lang.reflect.Array.newInstance(Node::class.java, DEFAULT_DECREASE_POOL_SIZE) as Array<Node<*, *>?>
        decreasePoolSize = 0
        decreasePoolMinPos = 0
    }

    /**
     * {@inheritDoc}
     */
    @LogarithmicTime(amortized = true)
    override fun deleteMin(): Node {
        if (size == 0L) {
            throw NoSuchElementException()
        }

        val min: Node?
        if (decreasePoolMinPos >= decreasePoolSize) {
            // decrease pool empty
            min = root

            // cut all children, and combine them
            root = combine(cutChildren(root))
        } else {
            val poolMin = decreasePool[decreasePoolMinPos.toInt()]
            val c: Int = comparator?.compare(root!!.key, poolMin!!.key) ?: (root!!.key as Comparable<K>?)!!.compareTo(poolMin!!.key)

            if (c <= 0) {
                // root is smaller
                min = root

                // cut children, combine
                val childrenTree = combine(cutChildren(root))
                root = null

                /*
                 * Append to decrease pool without updating minimum as we are
                 * going to consolidate anyway
                 */
                if (childrenTree != null) {
                    addPool(childrenTree, false)
                }
                consolidate()
            } else {
                // minimum in pool is smaller
                min = poolMin

                // cut children, combine
                val childrenTree = combine(cutChildren(poolMin))

                if (childrenTree != null) {
                    // add to location of previous minimum and consolidate
                    decreasePool[decreasePoolMinPos.toInt()] = childrenTree
                    childrenTree.poolIndex = decreasePoolMinPos
                } else {
                    decreasePool[decreasePoolMinPos.toInt()] = decreasePool[decreasePoolSize - 1]
                    decreasePool[decreasePoolMinPos.toInt()]!!.poolIndex = decreasePoolMinPos
                    decreasePool[decreasePoolSize - 1] = null
                    decreasePoolSize--
                }
                poolMin?.poolIndex = NO_INDEX

                consolidate()
            }
        }

        size--
        return min!!
    }

    /**
     * {@inheritDoc}
     *
     * This operation takes amortized zero cost.
     */
    @ConstantTime(amortized = true)
    override fun meld(other: MergeableAddressableHeap<K, V>) {
        val h = other as CostlessMeldPairingHeap<K, V>

        // check same comparator
        if (comparator != null) {
            require(!(h.comparator == null || h.comparator != comparator)) { "Cannot meld heaps using different comparators!" }
        } else require(h.comparator == null) { "Cannot meld heaps using different comparators!" }

        check(h.other === h) { "A heap cannot be used after a meld." }

        // meld
        if (size < h.size) {
            consolidate()

            root = if (comparator == null) {
                link(h.root, root)
            } else {
                linkWithComparator(h.root, root)
            }
            decreasePoolSize = h.decreasePoolSize
            h.decreasePoolSize = 0

            decreasePoolMinPos = h.decreasePoolMinPos
            h.decreasePoolMinPos = 0

            val tmp = decreasePool
            decreasePool = h.decreasePool
            h.decreasePool = tmp
        } else {
            h.consolidate()

            root = if (comparator == null) {
                link(h.root, root)
            } else {
                linkWithComparator(h.root, root)
            }
        }

        size += h.size
        h.root = null
        h.size = 0

        // take ownership
        h.other = this
    }

    // node
    inner class Node(/*
         * We maintain explicitly the belonging heap, instead of using an inner
         * class due to possible cascading melding.
         */var heap: CostlessMeldPairingHeap<K, V>,
                              override var key: K,
                              override var value: V?
    ) : AddressableHeap.Handle<K, V>, Serializable {
        /**
         * {@inheritDoc}
         */
        var o_c: Node? = null // older child
        var y_s: Node? = null // younger sibling
        var o_s: Node? = null // older sibling or parent
        var poolIndex: Byte = NO_INDEX // position in decrease pool

        /**
         * {@inheritDoc}
         */
        @LogLogTime(amortized = true)
        override fun delete() {
            owner.delete(this)
        }

        /**
         * {@inheritDoc}
         */
        @LogLogTime(amortized = true)
        override fun decreaseKey(newKey: K) {
            owner.decreaseKey(this, newKey)
        }

        val owner: CostlessMeldPairingHeap<K, V>
            /*
         * Get the owner heap of the handle. This is union-find with
         * path-compression between heaps.
         */
            get() {
                if (heap.other !== heap) {
                    // find root
                    var root = heap
                    while (root !== root.other) {
                        root = root.other
                    }
                    // path-compression
                    var cur = heap
                    while (cur.other !== root) {
                        val next = cur.other
                        cur.other = root
                        cur = next
                    }
                    heap = root
                }
                return heap
            }
    }

    /*
     * Decrease the key of a node.
     */
    private fun decreaseKey(n: Node, newKey: K) {
        var c: Int
        c = comparator?.compare(newKey, n.key) ?: (newKey as Comparable<K>).compareTo(n.key)

        require(c <= 0) { "Keys can only be decreased!" }
        n.key = newKey

        if (c == 0 || root === n) {
            // root or no change in key
            return
        } else // node has a parent

        // append node (minus oldest child) to decrease pool

        // if decrease pool has >= ceil(logn) trees, consolidate
// no parent and not root, so inside pool
            require(!(n.o_s == null && n.poolIndex == NO_INDEX)) {
                // no root, no parent and no pool index
                "Invalid handle!"
            }
        if (n.o_s == null) {
            // no parent and not root, so inside pool
            val poolMin = decreasePool[decreasePoolMinPos.toInt()]
            c = if (comparator == null) {
                (newKey as Comparable<K?>).compareTo(poolMin!!.key)
            } else {
                comparator.compare(newKey, poolMin!!.key)
            }
            if (c < 0) {
                decreasePoolMinPos = n.poolIndex
            }
            return
        } else {
            // node has a parent
            val oldestChild = cutOldestChild(n)
            if (oldestChild != null) {
                linkInPlace(oldestChild, n)
            } else {
                cutFromParent(n)
            }

            // append node (minus oldest child) to decrease pool
            addPool(n, true)

            // if decrease pool has >= ceil(logn) trees, consolidate
            val sizeAsDouble = size.toDouble()
            if (decreasePoolSize >= Math.getExponent(sizeAsDouble) + 1) {
                consolidate()
            }
        }
    }

    /**
     * Delete a node.
     *
     * @param n
     * the node
     */
    private fun delete(n: Node) {
        if (((n !== root) && n.o_s == null) && n.poolIndex == NO_INDEX) {
            // no root, no parent and no pool index
            throw IllegalArgumentException("Invalid handle!")
        }

        // node has a parent
        if (n.o_s != null) {
            // cut oldest child
            val oldestChild = cutOldestChild(n)
            if (oldestChild != null) {
                linkInPlace(oldestChild, n)
            } else {
                cutFromParent(n)
            }
        }

        // node has no parent
        // cut children
        val childrenTree = combine(cutChildren(n))
        var checkConsolidate = false
        if (childrenTree != null) {
            checkConsolidate = true
            addPool(childrenTree, true)
        }

        size--
        if (n === root) {
            root = null
            consolidate()
            checkConsolidate = false
        } else if (n.poolIndex != NO_INDEX) {
            val curIndex = n.poolIndex
            decreasePool[curIndex.toInt()] = decreasePool[decreasePoolSize - 1]
            decreasePool[curIndex.toInt()]!!.poolIndex = curIndex
            decreasePool[decreasePoolSize - 1] = null
            decreasePoolSize--
            n.poolIndex = NO_INDEX
            if (curIndex == decreasePoolMinPos) {
                // in decrease pool, and also the minimum
                consolidate()
                checkConsolidate = false
            } else {
                // in decrease pool, and not the minimum
                if (decreasePoolMinPos == decreasePoolSize) {
                    decreasePoolMinPos = curIndex
                }
                checkConsolidate = true
            }
        }

        // if decrease pool has >= ceil(logn) trees, consolidate
        if (checkConsolidate) {
            val sizeAsDouble = size.toDouble()
            if (decreasePoolSize >= Math.getExponent(sizeAsDouble) + 1) {
                consolidate()
            }
        }
    }

    /*
     * Consolidate. Combine the trees of the decrease pool in one tree by
     * sorting the values of the roots of these trees, and linking the trees in
     * this order such that their roots form a path of nodes in the combined
     * tree (make every root the leftmost child of the root with the next
     * smaller value). Join this combined tree with the main tree.
     */
    private fun consolidate() {
        if (decreasePoolSize.toInt() == 0) {
            return
        }

        // lazily initialize comparator
        if (decreasePoolComparator == null) {
            if (comparator == null) {
                decreasePoolComparator = Comparator { o1, o2 -> (o1?.key as Comparable<K?>).compareTo(o2?.key) }
            } else {
                decreasePoolComparator = Comparator { o1, o2 -> comparator.compare(o1?.key, o2?.key) }
            }
        }

        // sort
        //Arrays.sort(decreasePool, 0, decreasePoolSize.toInt(), decreasePoolComparator)

        decreasePoolComparator?.let { decreasePool.sortWith(it, 0, decreasePoolSize.toInt()) }

        var i = decreasePoolSize - 1
        var s = decreasePool[i]
        s!!.poolIndex = NO_INDEX
        while (i > 0) {
            val f = decreasePool[i - 1]
            f!!.poolIndex = NO_INDEX
            decreasePool[i] = null

            // link (no comparison, due to sort)
            s!!.y_s = f.o_c
            s.o_s = f
            if (f.o_c != null) {
                f.o_c!!.o_s = s
            }
            f.o_c = s

            // advance
            s = f
            i--
        }

        // empty decrease pool
        decreasePool[0] = null
        decreasePoolSize = 0
        decreasePoolMinPos = 0

        // merge tree with root
        root = if (comparator == null) {
            link(root!!, s)
        } else {
            linkWithComparator(root!!, s)
        }
    }

    /**
     * Append to decrease pool.
     */
    private fun addPool(n: Node, updateMinimum: Boolean) {
        decreasePool[decreasePoolSize.toInt()] = n
        n.poolIndex = decreasePoolSize
        decreasePoolSize++

        if (updateMinimum && decreasePoolSize > 1) {
            val poolMin = decreasePool[decreasePoolMinPos.toInt()]
            val c: Int
            c = if (comparator == null) {
                (n.key as Comparable<K>?)!!.compareTo(poolMin!!.key)
            } else {
                comparator.compare(n.key, poolMin!!.key)
            }
            if (c < 0) {
                decreasePoolMinPos = n.poolIndex
            }
        }
    }

    /**
     * Two pass pair and compute root.
     */
    private fun combine(l: Node?): Node? {
        if (l == null) {
            return null
        }

        assert(l.o_s == null)
        // left-right pass
        var pairs: Node? = null
        var it = l
        var p_it: Node?
        if (comparator == null) { // no comparator
            while (it != null) {
                p_it = it
                it = it.y_s

                if (it == null) {
                    // append last node to pair list
                    p_it.y_s = pairs
                    p_it.o_s = null
                    pairs = p_it
                } else {
                    val n_it = it.y_s

                    // disconnect both
                    p_it.y_s = null
                    p_it.o_s = null
                    it.y_s = null
                    it.o_s = null

                    // link trees
                    p_it = link(p_it, it)

                    // append to pair list
                    p_it?.y_s = pairs
                    pairs = p_it

                    // advance
                    it = n_it
                }
            }
        } else {
            while (it != null) {
                p_it = it
                it = it.y_s

                if (it == null) {
                    // append last node to pair list
                    p_it.y_s = pairs
                    p_it.o_s = null
                    pairs = p_it
                } else {
                    val n_it = it.y_s

                    // disconnect both
                    p_it.y_s = null
                    p_it.o_s = null
                    it.y_s = null
                    it.o_s = null

                    // link trees
                    p_it = linkWithComparator(p_it, it)

                    // append to pair list
                    p_it?.y_s = pairs
                    pairs = p_it

                    // advance
                    it = n_it
                }
            }
        }

        // second pass (reverse order - due to add first)
        it = pairs
        var f: Node? = null
        if (comparator == null) {
            while (it != null) {
                val nextIt = it.y_s
                it.y_s = null
                f = link(f, it)
                it = nextIt
            }
        } else {
            while (it != null) {
                val nextIt = it.y_s
                it.y_s = null
                f = linkWithComparator(f, it)
                it = nextIt
            }
        }

        return f
    }

    /**
     * Cut the children of a node and return the list.
     *
     * @param n
     * the node
     * @return the first node in the children list
     */
    private fun cutChildren(n: Node?): Node? {
        n?:return null
        val child = n.o_c
        n.o_c = null
        if (child != null) {
            child.o_s = null
        }
        return child
    }

    /**
     * Cut the oldest child of a node.
     *
     * @param n
     * the node
     * @return the oldest child of a node or null
     */
    private fun cutOldestChild(n: Node): Node? {
        val oldestChild = n.o_c
        if (oldestChild != null) {
            if (oldestChild.y_s != null) {
                oldestChild.y_s!!.o_s = n
            }
            n.o_c = oldestChild.y_s
            oldestChild.y_s = null
            oldestChild.o_s = null
        }
        return oldestChild
    }

    /**
     * Cut a node from its parent.
     *
     * @param n
     * the node
     */
    private fun cutFromParent(n: Node) {
        if (n.o_s != null) {
            if (n.y_s != null) {
                n.y_s!!.o_s = n.o_s
            }
            if (n.o_s!!.o_c === n) { // I am the oldest :(
                n.o_s!!.o_c = n.y_s
            } else { // I have an older sibling!
                n.o_s!!.y_s = n.y_s
            }
            n.y_s = null
            n.o_s = null
        }
    }

    /**
     * Put an orphan node into the position of another node. The other node
     * becomes an orphan.
     *
     * @param orphan
     * the orphan node
     * @param n
     * the node which will become an orphan
     */
    private fun linkInPlace(orphan: Node, n: Node) {
        // link orphan at node's position
        orphan.y_s = n.y_s
        if (n.y_s != null) {
            n.y_s!!.o_s = orphan
        }
        orphan.o_s = n.o_s
        if (n.o_s != null) {
            if (n.o_s!!.o_c === n) { // node is the oldest :(
                n.o_s!!.o_c = orphan
            } else { // node has an older sibling!
                n.o_s!!.y_s = orphan
            }
        }
        n.o_s = null
        n.y_s = null
    }

    private fun link(f: Node?, s: Node?): Node? {
        if (s == null) {
            return f
        } else if (f == null) {
            return s
        } else if ((f.key as Comparable<K>?)!!.compareTo(s.key) <= 0) {
            s.y_s = f.o_c
            s.o_s = f
            if (f.o_c != null) {
                f.o_c!!.o_s = s
            }
            f.o_c = s
            return f
        } else {
            return link(s, f)
        }
    }

    private fun linkWithComparator(f: Node?, s: Node?): Node? {
        if (s == null) {
            return f
        } else if (f == null) {
            return s
        } else if (comparator!!.compare(f.key, s.key) <= 0) {
            s.y_s = f.o_c
            s.o_s = f
            if (f.o_c != null) {
                f.o_c!!.o_s = s
            }
            f.o_c = s
            return f
        } else {
            return linkWithComparator(s, f)
        }
    }

    companion object {
        const val NO_INDEX: Byte = -1
        
        /**
         * Maximum length of decrease pool for long type.
         */
        private const val DEFAULT_DECREASE_POOL_SIZE = 64 + 1
    }
}
